package com.gxlevi.db.util

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.util.Bytes
import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}


/**
 * HBase工具类
 *
 * 获取Table
 * 保存单列数据
 * 查询单列数据
 * 保存多列数据
 * 查询多列数据
 * 删除数据
 */
object HBaseUtil {
  //HBase的配置类,不需要指定配置文件,文件名要求hbase-site.xml
  val conf: Configuration = HBaseConfiguration.create()
  //HBase的连接
  val conn: Connection = ConnectionFactory.createConnection(conf)
  //HBase的操作api
  val admin: Admin = conn.getAdmin

  /**
   * 返回table 如果不存在,则创建表
   *
   * @param tableNameStr
   * @param columnFamilyNameStr
   * @return
   */
  def getTable(tableNameStr: String, columnFamilyNameStr: String): Table = {
    //获取tableName
    val tableName = TableName.valueOf(tableNameStr)
    //表不存在则创建表
    if (!admin.tableExists(tableName)) {
      //构建出 表的描述的建造者
      val descBuilder: TableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableName)
      val familyDescriptor: ColumnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(columnFamilyNameStr.getBytes()).build()
      //给表添加列族
      descBuilder.setColumnFamily(familyDescriptor)

      //创建表
      admin.createTable(descBuilder.build())
    }
    conn.getTable(tableName)
  }

  /**
   *
   * @param tableName        表名
   * @param rowKey           rowkey
   * @param columnFamilyName 列族名
   * @param columnName       列名
   * @param columnValue      列值
   */
  def putData(tableName: String, rowKey: String, columnFamilyName: String, columnName: String, columnValue: String) = {
    //获取表
    val table: Table = getTable(tableName, columnFamilyName)
    try {
      //Put
      val put = new Put(rowKey.getBytes())
      put.addColumn(columnFamilyName.getBytes(), columnName.getBytes(), columnValue.getBytes())
      //保存数据
      table.put(put)
    } catch {
      case ex: Exception => {
        ex.printStackTrace()
      }
    } finally {
      table.close()
    }
  }

  /**
   * 通过单列名获取列值
   *
   * @param tableNameStr     表名
   * @param rowkey           rowkey
   * @param columnFamilyName 列族名
   * @param columnName       列名
   * @return 列值
   */
  def getData(tableNameStr: String, rowkey: String, columnFamilyName: String, columnName: String): String = {
    //1.获取table对象
    val table = getTable(tableNameStr, columnFamilyName)
    try {
      //2.构建Get对象
      val get = new Get(rowkey.getBytes())
      //3.进行查询
      val result: Result = table.get(get)
      //4.判断结果是否为空,并且包含要查询的列
      if (result != null && result.containsColumn(columnFamilyName.getBytes(), columnName.getBytes())) {
        val bytes = result.getValue(columnFamilyName.getBytes(), columnName.getBytes())
        Bytes.toString(bytes)
      } else {
        ""
      }
    } catch {
      case ex: Exception => ex.printStackTrace()
        ""
    } finally {
      //5.关闭表
      table.close()
    }
  }

  /**
   * 存储多列数据
   *
   * @param tableNameStr     表名
   * @param rowkey           rowkey
   * @param columnFamilyName 列族名
   * @param map              多个列名列值集合
   */
  def putMapData(tableNameStr: String, rowkey: String, columnFamilyName: String, map: Map[String, Any]) = {
    //1.获取Table
    val table: Table = getTable(tableNameStr, columnFamilyName)
    try {
      //2.创建put
      val put = new Put(rowkey.getBytes())
      //3.在put中添加多个列名和列值
      for ((colName, colValue) <- map) {
        put.addColumn(columnFamilyName.getBytes(), colName.getBytes(), colValue.toString.getBytes())
      }
      //4.保存put
      table.put(put)
    } catch {
      case ex: Exception => ex.printStackTrace()
    } finally {
      //5.关闭table
      table.close()
    }
  }

  /**
   * 获取多列数据的值
   *
   * @param tableNameStr     表名
   * @param rowkey           rowkey
   * @param columnFamilyName 列族名
   * @param columnNameList   多个列名
   * @return 多个列名和列值
   */
  def getMapData(tableNameStr: String, rowkey: String, columnFamilyName: String, columnNameList: List[String]): Map[String, String] = {
    //1.获取table
    val table = getTable(tableNameStr, columnFamilyName)
    try {
      //2.构建get
      val get = new Get(rowkey.getBytes())
      //3.执行查询
      val result = table.get(get)
      //4.遍历列名集合,取出列值,构建成map返回
      columnNameList.map {
        col =>
          val bytes: Array[Byte] = result.getValue(columnFamilyName.getBytes(), col.getBytes())
          if (bytes != null && bytes.size > 0) {
            col -> Bytes.toString(bytes)
          } else {
            "" -> ""
          }
      }.filter(_._1 != "").toMap
    } catch {
      case ex: Exception => {
        ex.printStackTrace()
        Map[String, String]()
      }
    } finally {
      //关闭table
      table.close()
    }
  }

  /**
   * 删除数据
   *
   * @param tableNameStr
   * @param rowkey
   * @param columnFamilyName
   */
  def deleteData(tableNameStr: String, rowkey: String, columnFamilyName: String) = {
    //1.获取table
    val table = getTable(tableNameStr, columnFamilyName)
    try {
      //2.构建delete对象
      val delete = new Delete(rowkey.getBytes())
      //3.执行删除
      table.delete(delete)
    } catch {
      case ex: Exception => ex.printStackTrace()
    } finally {
      //4.关闭table
      table.close()
    }
  }

  def main(args: Array[String]): Unit = {
    //    println(getTable("test", "info"))
    //    putData("test", "1", "info", "t1", "hello world")
    /*    val str = getData("test", "1", "info", "t1")
        println(str)*/
    /*    val map = Map(
          "t2" -> "scala",
          "t3" -> "hive",
          "t4" -> "flink"
        )
        putMapData("test", "1", "info", map)*/

    deleteData("test", "1", "info")
    println(getMapData("test", "1", "info", List("t1", "t2")))

  }
}
